package com.thiswhyme.republish.rabbitmq.core;

import com.thiswhyme.republish.rabbitmq.config.BindingPlusProperties;
import com.thiswhyme.republish.rabbitmq.config.BindingServicePlusProperties;
import com.thiswhyme.republish.rabbitmq.core.domain.MessageInfo;
import com.thiswhyme.republish.rabbitmq.core.domain.PublishMqHeaders;
import com.thiswhyme.republish.rabbitmq.core.domain.PublishStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.util.CollectionUtils;

import java.util.List;

/**
 * @description: 消息发送确认处理器-重发处理器抽象类
 * @author: thiswhyme
 * @date: 2021/1/15
 */
public abstract class AbstractRepublishHandler implements IPublishAckHandler {

    public static final Logger log = LoggerFactory.getLogger(AbstractRepublishHandler.class);

    @Autowired
    private MessageInfoService messageInfoService;

    @Autowired
    private RepublishRedisOperation redisOperation;

    @Autowired
    private BindingServicePlusProperties bindingServicePlusProperties;

    /**
     * 消息未路由到queue处理
     * 生产环境一般不会出现，开发过程中可能存在路由key配置错误问题，所以此处提供默认处理：打印日志
     * @param message
     */
    @Override
    public void routeFail(Message<?> message) {
        log.error("【发送确认处理器】-消息未路由到queue，消息UUID：{}, 内容：{}", message.getHeaders().getId(), message);
    }

    /**
     * 发送端confirm ack处理（配合重发）
     * @param message 消息
     */
    protected void handleConfirmForRepublish(Message<?> message) {
        try {
            //校验
            boolean ack = (Boolean) message.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM);
            Object uniqueIdObj = message.getHeaders().get(PublishMqHeaders.HEADER_RETRY_UNIQUE_ID);
            if(null == uniqueIdObj || CollectionUtils.isEmpty(bindingServicePlusProperties.getBindings())) {
                log.debug("【发送确认处理器】-收到消息ack，未获取到消息ID，跳过处理，消息UUID：{}，ack：{}",
                        message.getHeaders().getId().toString(), ack);
                return;
            }
            Long uniqueId = (Long) uniqueIdObj;
            MessageInfo messageInfo = messageInfoService.queryById(uniqueId);
            if(null == messageInfo) {
                log.info("【发送确认处理器】-收到消息ack，未查询到消息内容，跳过处理，消息ID：{}，ack：{}", uniqueId, ack);
                return;
            }
            BindingPlusProperties properties = bindingServicePlusProperties.getBindings().get(messageInfo.getChannelName());
            if(null == properties || !properties.getEnableRepublish()) {
                log.info("【发送确认处理器】-收到消息ack，未开启重试，跳过处理，消息ID：{}，ack：{}", uniqueId, ack);
                return;
            }

            if(ack) {
                doAckSuccess(uniqueId);
            } else {
                doAckFail(message, uniqueId, messageInfo, properties);
            }
        } catch (Exception e) {
            Object uniqueIdObj = message.getHeaders().get(PublishMqHeaders.HEADER_RETRY_UNIQUE_ID);
            log.error("【发送确认处理器】-收到消息ack，处理ack消息失败，消息ID：{}", uniqueIdObj, e);
        }
    }

    /**
     * ack成功处理
     * @param uniqueId
     */
    private void doAckSuccess(Long uniqueId) {
        log.debug("【发送确认处理器】-ack成功，从待ack和待republish的zset剔除，更新数据库状态，消息ID：{}", uniqueId);
        redisOperation.removeFromAllZSet(uniqueId);
        messageInfoService.updatePublishStatus(uniqueId, PublishStatus.SEND_SUCCESS);
    }

    /**
     * ack失败处理
     * @param message
     * @param uniqueId
     * @param messageInfo
     * @param properties
     */
    private void doAckFail(Message<?> message, Long uniqueId, MessageInfo messageInfo, BindingPlusProperties properties) {
        int retryLimit = properties.getRepublishTimes();
        int current = (int) message.getHeaders().get(PublishMqHeaders.HEADER_SEND_RETRY_TIMES);
        if(current < retryLimit && messageInfo.getPublishStatus() == PublishStatus.SEND_TO_ACK.getValue()) {
            log.info("【发送确认处理器】-ack失败，置入待重发的zset，准备第{}次重发，消息ID：{}", current + 1, uniqueId);
            List<Integer> delayMillisList = properties.getDelayRepublishMillis();
            int delayMillis = delayMillisList.get(current > delayMillisList.size() ? delayMillisList.size() - 1 : current);
            redisOperation.moveToRepublishSet(uniqueId,  System.currentTimeMillis() + delayMillis);
        } else {
            log.info("【发送确认处理器】-ack失败，重发{}次仍然失败，更新数据库状态，消息ID：{}", current, uniqueId);
            messageInfoService.updatePublishStatus(uniqueId, PublishStatus.FAIL);
            doBizRepublishFail(message, messageInfo.getChannelName(), "消息重发指定次数后仍然失败");
        }
    }

    /**
     * 重发失败的自定义业务处理
     * @param message 消息内容
     * @param outPutChannelName 发送通道名
     * @param errMsg 错误信息
     */
    public abstract void doBizRepublishFail(Message<?> message, String outPutChannelName, String errMsg);
}
